package com.wyd.hander;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
@Slf4j
public class KafkaConsumer {

//  @KafkaListener(topics = {"${kafka.topic.order}"})
  public void receiveOrder(ConsumerRecord<String, String> record) {
      String topic = record.topic();
      String msg = record.value();
      log.info("消费者接受消息：topic-->"+topic+",msg->>"+msg);
  }

    @KafkaListener(containerFactory = "manualImmediateListenerContainerFactory" , topics = {"${kafka.topic.order}"})
    public void onMessageManual(List<Object> message, Acknowledgment ack){
        log.info("处理数据量：{}",message.size());
        message.forEach(item -> log.info("处理数据内容：{}",item));
        //直接提交offset
        ack.acknowledge();
    }

}